from flask import Flask, render_template, jsonify
from confluent_kafka import Consumer, KafkaException, KafkaError
import threading
import logging
from collections import defaultdict

app = Flask(__name__)

# 初始化统计变量
valid_count = 0
invalid_count = 0
product_valid_invalid_counts = defaultdict(lambda: {'Valid': 0, 'Invalid': 0})
category_counts = defaultdict(int)
category_product_valid_invalid_counts = defaultdict(lambda: defaultdict(lambda: {'Valid': 0, 'Invalid': 0}))

# 配置Kafka消费者
conf = {
    'bootstrap.servers': '192.168.192.138:9092',
    'group.id': 'niit',
    'auto.offset.reset': 'latest'
}

consumer = Consumer(conf)
consumer.subscribe(
    ['order_valid_invalid_count', 'product_order_stats', 'category_order_counts', 'category_valid_invalid_counts',
     'category_product_order_stats'])

# 设置日志记录器
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

lock = threading.Lock()

def consume_messages():
    global valid_count, invalid_count, product_valid_invalid_counts, category_counts, category_product_valid_invalid_counts
    try:
        while True:
            msg = consumer.poll(timeout=1.0)  # 轮询消息
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    logging.info(f"End of partition reached {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")
                else:
                    raise KafkaException(msg.error())
            else:
                value = msg.value().decode('utf-8')
                logging.info(f"接收信息 topic {msg.topic()}: {value}")
                try:
                    with lock:
                        if msg.topic() == 'order_valid_invalid_count':
                            isValid, count = value.split('\t'
                                                         '')
                            count = int(count.strip())
                            if isValid.strip() == 'Y':
                                valid_count += count
                            elif isValid.strip() == 'N':
                                invalid_count += count
                            logging.info(f"Updated {isValid.strip()} count: {count}")
                        elif msg.topic() == 'product_order_stats':
                            productName, isValid, count = value.split('\t')
                            count = int(count.strip())
                            isValid = 'Valid' if isValid.strip() == 'Y' else 'Invalid'
                            product_valid_invalid_counts[productName.strip()][isValid] += count
                            logging.info(f"Updated {productName.strip()} {isValid} count: {count}")
                        elif msg.topic() == 'category_order_counts':
                            category, count = value.split('\t')
                            count = int(count.strip())
                            category_counts[category.strip()] += count
                            logging.info(f"Updated {category.strip()} count: {count}")

                        elif msg.topic() == 'category_product_order_stats':
                            category, productName, isValid, count = value.split('\t')
                            count = int(count.strip())
                            isValid = 'Valid' if isValid.strip() == 'Y' else 'Invalid'
                            category_product_valid_invalid_counts[category.strip()][productName.strip()][isValid] += count
                            logging.info(f"Updated {category.strip()} {productName.strip()} {isValid} count: {count}")
                except ValueError as ve:
                    logging.error(f"数据错误: {value}. Error: {ve}")
                except Exception as e:
                    logging.error(f"意外错误: {value}. Error: {e}")
    except KeyboardInterrupt:
        pass
    finally:
        consumer.close()

# 启动一个线程来持续消费消息
threading.Thread(target=consume_messages, daemon=True).start()

@app.route('/')
def index():
    return render_template('index.html')
@app.route('/get_all_stats')
def get_all_stats():
    global valid_count, invalid_count, product_valid_invalid_counts, category_counts, category_product_valid_invalid_counts
    with lock:
        stats = {
            'valid_count': valid_count,
            'invalid_count': invalid_count,
            'product_valid_invalid_counts': dict(product_valid_invalid_counts),
            'category_counts': dict(category_counts),
            'category_product_valid_invalid_counts': {k: dict(v) for k, v in category_product_valid_invalid_counts.items()}
        }
        logging.info(f"发送数据到前端: {stats}")
    return jsonify(stats)


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)